package com.wanke.config;

import com.wanke.domain.po.Measurement;
import com.wanke.domain.po.msg.CloudMeasureValue;
import com.wanke.domain.po.msg.CloudMeasureValues;
import com.wanke.subscribe.MqttMeasuremenSubscribetRecevice;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;

public class MeasurementCacheConfig {

    /**
     * 缓存数据
     */
    public  int cacheSize = 102400;


    private Long stationCode;

    /**
     * 测点数据缓存。</br>
     * 实时测点数据插入队列头部，非实时测点数据插入队列尾部。
     */
    LinkedBlockingDeque<Measurement> measureValueQueueForTransmit = new LinkedBlockingDeque<>(cacheSize);

    LinkedBlockingQueue<Measurement> measureValueQueueForStore = new LinkedBlockingQueue<>(cacheSize);

    private final Object measureValueUpdatedLock = new Object();

    private static final Logger logger = LoggerFactory.getLogger(MqttMeasuremenSubscribetRecevice.class);

    /**
     * 插入数据
     */
    public void insert(Measurement value) {
        boolean offerSuccess;
        //插入到发送缓存
        if ((value.getFlag() & 0x01) > 0) {
            // 是实时测点
            offerSuccess = measureValueQueueForTransmit.offerFirst(value);
            synchronized (measureValueUpdatedLock) {
                measureValueUpdatedLock.notifyAll();
            }
        } else {
            //非实时测点
            offerSuccess = measureValueQueueForTransmit.offerLast(value);
            synchronized (measureValueUpdatedLock) {
                measureValueUpdatedLock.notifyAll();
            }
        }
        if (!offerSuccess) {
            logger.warn("transmit cache[size={}] is full, measure has been discarded", measureValueQueueForTransmit.size());
        }
        measureValueQueueForStore.offer(value);
        logger.debug("接收点号队列长度:{}", measureValueQueueForTransmit.size());
    }


    /**
     * 从Queue中提取数据，生成与云端通信的消息对象
     *
     * @param count               提取数据的最大数量。
     * @return
     */
    public synchronized CloudMeasureValues extractCloudMeasureValues(int count, Long stationCode) {
        CloudMeasureValues cloudMeasureValues = new CloudMeasureValues(stationCode);
        List<CloudMeasureValue> values = new ArrayList<>(count);
        cloudMeasureValues.setValuesList(values);
        for (int i = 0; i < count; i++) {
            if (measureValueQueueForTransmit.isEmpty()) {
                break;
            }
            Measurement po = measureValueQueueForTransmit.poll();
            if (po != null) {
                values.add(po.toCloudMeasureValue());
            }
        }
        return cloudMeasureValues;
    }

    /**
     * 从Queue中提取数据，生成与云端通信的消息对象
     *
     * @param count               提取数据的最大数量。
     * @return
     */
    public synchronized List<Measurement> extractCloudMeasurement(int count) {
        List<Measurement> values = new ArrayList<>(count);
        for (int i = 0; i < count; i++) {
            if (measureValueQueueForTransmit.isEmpty()) {
                break;
            }
            Measurement po = measureValueQueueForTransmit.poll();
            if (po != null) {
                values.add(po);
            }
        }
        return values;
    }

    /**
     * 获得缓存大小
     */
    public int getSize() {
        return measureValueQueueForTransmit.size();
    }

    public Measurement peek() {
        return measureValueQueueForTransmit.peek();
    }
    public static final MeasurementCacheConfig INSTANCE = new MeasurementCacheConfig();
}
